package master

import (
	"bufio"
	"encoding/json"
	"fmt"
	"mapreduce-go/common"
	"os"
	"sort"
)

// 将多个reduce节点生成的最终结果输出文件输出到一个文件中进行排序
// 需要的参数
// 节点数量、任务名称
func (mr *Master) merge() {
	fmt.Println("merge result")

	kvs := make(map[string][]string)

	// 外层循环遍历所有文件
	for i := 0; i < mr.NReduce; i++ {
		filename := common.GetMergeName(mr.JobName, i)

		file, err := os.Open(filename)
		if err != nil {
			fmt.Println("error : ", err.Error())
		}

		decoder := json.NewDecoder(file)

		// 内层循环遍历文件中的每一行
		for decoder.More() {
			var kv common.KeyValue

			err = decoder.Decode(&kv)
			if err != nil {
				fmt.Println("error : ", err.Error())
			}

			// 将具有相同key的内容进行合并
			kvs[kv.Key] = append(kvs[kv.Key], kv.Value)
		}

		file.Close()
	}

	// 取出所有的key 以便于排序
	var keys []string
	for k, _ := range kvs {
		keys = append(keys, k)
	}

	// 排序
	sort.Strings(keys)

	// 创建一个最终的输出结果文件
	file, err := os.Create("mrtmp." + mr.JobName)
	if err != nil {
		fmt.Println("error : ", err.Error())
	}

	// 创建一个写入器
	writer := bufio.NewWriter(file)

	for _, k := range keys {
		fmt.Fprintf(writer, "%s: %s\n", k, kvs[k])
	}

	writer.Flush()
	file.Close()
}
